package com.haibara.codesubmiteval.kafka.listener;

import cn.hutool.json.JSONUtil;
import com.haibara.codesubmiteval.constant.KafkaConstant;
import com.haibara.codesubmiteval.judge.model.ExecuteCodeRespond;
import com.haibara.codesubmiteval.judge.service.JudgeService;
import com.haibara.codesubmiteval.model.entity.QuestionSubmit;
import com.haibara.codesubmiteval.model.enums.JudgeResultEnum;
import com.haibara.codesubmiteval.model.enums.QuestionSubmitStatusEnum;
import com.haibara.codesubmiteval.service.QuestionSubmitService;
import jakarta.annotation.Resource;
import lombok.extern.slf4j.Slf4j;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.retry.support.RetryTemplate;
import org.springframework.stereotype.Component;

/**
 * @Author: mengyu
 * @Date: 2025/2/10
 */
@Component
@Slf4j
public class JudgeListener {
    @Resource
    private RetryTemplate kafkaRetryTemplate;
    @Resource
    private JudgeService judgeService;
    @Resource
    private QuestionSubmitService questionSubmitService;

    @KafkaListener(topics = KafkaConstant.JUDGE_TOPIC, groupId = KafkaConstant.JUDGE_GROUP)
    public void consumerJudgeMessage(String content, Acknowledgment ack) {
        ExecuteCodeRespond ecr = JSONUtil.toBean(content, ExecuteCodeRespond.class);
        log.info("接收到判题请求:{}", ecr);
        kafkaRetryTemplate.execute(retryContext -> {
            try {
                judgeService.doJudge(ecr);
                ack.acknowledge();
            } catch (Exception e) {
                int retryCount = retryContext.getRetryCount();
                log.error("判题请求处理报错，重试次数:{}，题目提交id:{}", retryCount, ecr.getQuestionSubmitId());
                throw e;
            }
            return null;
        }, retryContext -> {
            Long questionSubmitId = ecr.getQuestionSubmitId();
            QuestionSubmit questionSubmit = questionSubmitService.getById(questionSubmitId);
            questionSubmit.setStatus(QuestionSubmitStatusEnum.FAILED.getValue());
            questionSubmit.setTimeUsed(0);
            questionSubmit.setMemoryUsed(0);
            questionSubmit.setJudgeResult(JudgeResultEnum.SYSTEM_ERROR.getText());
            log.error("判题失败 : {}", questionSubmit);
            ack.acknowledge();
            return null;
        });
    }
}
